package com.ht.shortlink.project.mq.consumer;

import cn.hutool.core.lang.UUID;
import com.ht.shortlink.project.common.convention.exception.ServiceException;
import com.ht.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.ht.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import com.ht.shortlink.project.service.IShortLinkService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executors;
import java.util.concurrent.locks.LockSupport;

import static com.ht.shortlink.project.common.constant.RedisCacheConstant.DELAY_QUEUE_STATS_KEY;

/**
 *  延迟队列记录统计数据组件
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class DelayShortLinkStatsConsumer implements InitializingBean {
    private final RedissonClient redissonClient;
    private final IShortLinkService shortLinkService;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;
    public void onMessage(){
        Executors.newSingleThreadExecutor(r -> {
                    Thread thread = new Thread();
                    thread.setName("delay_short_link_stats_consumer");
                    thread.setDaemon(Boolean.TRUE);
                    return thread;
                })
                .execute(() -> {
                    RBlockingDeque<ShortLinkStatsRecordDTO> blockingDeque = redissonClient.getBlockingDeque(DELAY_QUEUE_STATS_KEY);
                    RDelayedQueue<ShortLinkStatsRecordDTO> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
                    while (true) {
                        try {
                            ShortLinkStatsRecordDTO statsRecord = delayedQueue.poll();
                            if(statsRecord != null) {
                                try {
                                    statsRecord.setKeys(UUID.fastUUID().toString());
                                    if(messageQueueIdempotentHandler.isMessageQueueProcessed(statsRecord.getKeys())){
                                        if(messageQueueIdempotentHandler.isMessageQueueAccomplished(statsRecord.getKeys())){
                                            return;
                                        }
                                        throw new ServiceException("消息未完成流程，需要重试");
                                    }
                                    shortLinkService.shortLinkStats(null, statsRecord);
                                    continue;
                                } catch (Throwable ex) {
                                    log.error("消息统计出现异常", ex);
                                    messageQueueIdempotentHandler.deleteMessageQueueProcessedKey(statsRecord.getKeys());
                                }
                                messageQueueIdempotentHandler.setMessageQueueAccomplished(statsRecord.getKeys());
                            }
                            LockSupport.parkUntil(500);
                        } catch (Throwable ignored){

                        }
                    }
        });
    }
    @Override
    public void afterPropertiesSet() throws Exception {
        onMessage();
    }
}
